package collection.blocking.queue;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import collection.blocking.queue.exception.BlockingQueueIsClosedException;

public class BlockingQueue<T> {

    private final int capacity;
    private List<T> list;
    private int size;
    private boolean closing;

    private final Lock lock = new ReentrantLock();
    private final Condition notEmpty;
    private final Condition notFull;

    public BlockingQueue(final int capacity) {
        if (capacity <= 0) {
            throw new IllegalArgumentException();
        }
        this.capacity = capacity;
        this.size = 0;
        list = new LinkedList<T>();
        closing = false;
        notEmpty = lock.newCondition();
        notFull = lock.newCondition();
    }

    public T take() throws InterruptedException,
            BlockingQueueIsClosedException {
        lock.lock();
        try {
            while ((!closing) && (size == 0)) {
                notFull.await();
            }
            if (closing) {
                throw new BlockingQueueIsClosedException();
            }
            T item = list.remove(0);
            --size;
            notEmpty.signalAll();
            return item;
        } finally {
            lock.unlock();
        }
    }

    public void put(final T item) throws InterruptedException,
            BlockingQueueIsClosedException {
        if (item == null) {
            throw new NullPointerException();
        }
        lock.lock();
        try {
            while ((!closing) && (size >= capacity)) {
                notEmpty.await();
            }
            if (closing) {
                throw new BlockingQueueIsClosedException();
            }
            list.add(item);
            ++size;
            notFull.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public boolean isEmpty() {
        lock.lock();
        try {
            return size == 0;
        } finally {
            lock.unlock();
        }
    }

    public boolean isFull() {
        lock.lock();
        try {
            return size == capacity;
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        return size;
    }

    public void close() {
        lock.lock();
        try {
            closing = true;
            notFull.signalAll();
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public boolean isClosed() {
        return closing;
    }
}